package cn.cupster.flowdemo.fragment

import android.os.Bundle
import android.view.LayoutInflater
import android.view.View
import android.view.ViewGroup
import android.widget.TextView
import androidx.fragment.app.Fragment
import androidx.lifecycle.lifecycleScope
import cn.cupster.flowdemo.R
import cn.cupster.flowdemo.bean.Vp2PageBean
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.*
import java.lang.RuntimeException


class Vp2PageFragment : Fragment() {

    companion object {

        const val KEY_ARGUMENT_PAGE_BEAN = "BEAN"

        fun create(pageBean: Vp2PageBean?): Fragment {
            return Vp2PageFragment().apply {
                arguments = Bundle().apply {
                    putSerializable(KEY_ARGUMENT_PAGE_BEAN, pageBean)
                }
            }
        }
    }

    private lateinit var mTvCodeStr: TextView
    private lateinit var mTvResult: TextView
    private lateinit var mBtnExe: TextView
    private lateinit var mBtnClean: TextView
    private lateinit var mPageBean: Vp2PageBean

    override fun onCreateView(
        inflater: LayoutInflater,
        container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
        val fragLayout = inflater.inflate(R.layout.fragment_vp2_page, container, false)
        mTvCodeStr = fragLayout.findViewById(R.id.tvCodeStr)
        mTvResult = fragLayout.findViewById(R.id.tvResult)
        mBtnExe = fragLayout.findViewById(R.id.btnExe)
        mBtnClean = fragLayout.findViewById(R.id.btnClean)
        return fragLayout
    }

    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        super.onViewCreated(view, savedInstanceState)
        val pageBean: Vp2PageBean? =
            arguments?.getSerializable(KEY_ARGUMENT_PAGE_BEAN) as? Vp2PageBean
        updateData(pageBean)
    }

    fun updateData(pageBean: Vp2PageBean?) {
        pageBean?.let {
            mPageBean = it
            mTvCodeStr.text = it.codeStr
            mTvResult.append("${it.descStr}\n")
            mBtnClean.setOnClickListener {
                mTvResult.text = ""
            }
            mBtnExe.setOnClickListener { clickExecute(it) }
        }
    }

    private fun clickExecute(view: View) {
        when (mPageBean.action) {
            Vp2PageBean.ACTION_BASE -> {
                //1.简单使用
                exeFlowBase()
            }
            Vp2PageBean.ACTION_FLOW_OF -> {
                //2.list快捷转换flow
                exeFlowOf()
            }
            Vp2PageBean.ACTION_AS_FLOW -> {
                //3.list快速转换flow
                exeAsFlow()
            }
            Vp2PageBean.ACTION_CHANNEL_FLOW -> {
                //4.使用channelFlow
                exeChannelFlow()
            }
            Vp2PageBean.ACTION_FLOW_ON -> {
                //5.使用flow切换线程
                exeFlowOn()
            }
            Vp2PageBean.ACTION_CATCH -> {
                //6.异常处理catch{}
                exeFlowCatch()
            }
            Vp2PageBean.ACTION_CATCH_2 -> {
                //7.异常处理catch AfterCompletion
                exeFlowCatch2()
            }
            Vp2PageBean.ACTION_RETRY -> {
                //8.异常重试retry
                exeFlowRetry()
            }
            Vp2PageBean.ACTION_BUFFER -> {
                //9.exeBuffer-背压缓存策略
                exeBuffer()
            }
            Vp2PageBean.ACTION_CONFLATE -> {
                //10.conflate-合并-背压缓存策略
                exeConflate()
            }
            Vp2PageBean.ACTION_FLATE_MAP_MERGE -> {
                //11.exeFlatMapMerge并行执行
                exeFlatMapMerge()
            }
            Vp2PageBean.ACTION_FLATE_MAP_LATEST -> {
                //12.exeFlatMapMerge并行执行
                exeFlatMapLatest()
            }
            Vp2PageBean.ACTION_FLATE_MAP_CONTACT -> {
                //13.exeFlatMapMerge并行执行
                exeFlatMapConcat()
            }
            Vp2PageBean.ACTION_FLATTEN_MERGE -> {
                //14.exeFlatMapMerge并行执行
                exeFlattenMerge()
            }
            Vp2PageBean.ACTION_COMBINE -> {
                //15.exeCombine
                exeCombine()
            }
            Vp2PageBean.ACTION_ZIP -> {
                //16.exeZip
                exeZip()
            }
            Vp2PageBean.ACTION_FOLD -> {
                //17.exeFold 折叠、累积操作，可设置初始值
                exeFold()
            }
            Vp2PageBean.ACTION_REDUCE -> {
                //18.exeReduce 合并、累积操作
                exeReduce()
            }
            Vp2PageBean.ACTION_TAKE -> {
                //19.exeTake  只取前面几个emit发射的值
                exeTake()
            }
            Vp2PageBean.ACTION_TRANSFORM -> {
                //19.exeTransform  可以多次调用emit
                exeTransform()
            }
        }
    }

    /**
     * 1.简单使用
     */
    private fun exeFlowBase() {
        lifecycleScope.launch {
            flow<Int> {
                for (num in 1..5) {
                    delay(250)
                    emit(num)//发射数据
                }
            }.onCompletion {
                mTvResult.append("==========\n")
            }.collect {//等价于订阅操作
                mTvResult.append("$it\n")
            }
        }
    }

    /**
     * 2.list快捷转换flow
     */
    private fun exeFlowOf() {
        lifecycleScope.launch {
            flowOf(1, 2, 3, 4, 5)
                .onEach {
                    delay(250)
                }.onCompletion {
                    mTvResult.append("==========\n")
                }.collect {//等价于订阅操作
                    mTvResult.append("$it\n")
                }
        }
    }

    /**
     * 3.list快速转换flow
     */
    private fun exeAsFlow() {
        lifecycleScope.launch {
            mutableListOf(1, 2, 3, 4, 5).asFlow()
                .onEach {
                    delay(250)
                }.onCompletion {
                    mTvResult.append("==========\n")
                }.collect {//等价于订阅操作
                    mTvResult.append("$it\n")
                }
        }
    }

    /**
     * 4.使用channelFlow
     */
    private fun exeChannelFlow() {
        lifecycleScope.launch {
            //channelFlow是热数据流，flow是冷数据流
            //channelFlow默认实现异步非阻塞，flow需自行切换线程才能实现
            channelFlow<Int> {
                for (num in 1..5) {
                    delay(250)
                    send(num)//发射数据(异步非阻塞)
                }
                awaitClose {  }
            }.onCompletion {
                mTvResult.append("==========\n")
            }.collect {//等价于订阅操作
                mTvResult.append("$it\n")
            }
        }
    }

    /**
     * 5.使用flow切换线程
     */
    private fun exeFlowOn() {
        lifecycleScope.launch {
            flow<Int> {
                for (num in 1..5) {
                    delay(250)
                    emit(num)//发射数据
                }
            }
//                .flowOn(Dispatchers.IO)//切换线程，影响前面代码的执行线程，不影响后面
                .onCompletion {// 执行线程会被 flowOn() 改变，map()、each{}同理
                    val completionThread = Thread.currentThread().name
                    mTvResult.post {// 非主线程不能UI更新操作,不能使用 withContext() 来切换 flow 的线程
                        mTvResult.append("${completionThread}:===== end =====\n")
                    }
                }
                .flowOn(Dispatchers.IO)//切换线程，影响前面代码的执行线程
                .collect {//collect的执行线程，取决于执行协程的Scope
                    mTvResult.append("${Thread.currentThread().name}:$it\n")
                }
        }
    }

    /**
     * 6.异常处理catch{}
     */
    private fun exeFlowCatch() {
        lifecycleScope.launch {
            (1..4).asFlow()
                .onEach {
                    delay(250)
                    if (it == 3) {
                        throw RuntimeException()
                    }
                }.catch {//catch的调用位置决定异常捕获范围
                    val completionThread = Thread.currentThread().name
                    mTvResult.post {
                        mTvResult.append("${completionThread}:catch exception\n")
                    }
                }.onCompletion { cause ->
                    val completionThread = Thread.currentThread().name
                    if (cause != null) {//catch{}位置在前，异常已经被消费，这里捕获不到，为null,执行else
                        mTvResult.post {
                            mTvResult.append("${completionThread}:end with exception\n")
                        }
                    } else {
                        mTvResult.post {
                            mTvResult.append("${completionThread}:end normal\n")
                        }
                    }
                }
//                .flowOn(Dispatchers.IO)
                .collect {//订阅，flow 才会执行
                    mTvResult.append("${Thread.currentThread().name}:$it\n")
                }
        }
    }

    /**
     * 7.异常处理catch AfterCompletion
     */
    private fun exeFlowCatch2() {
        lifecycleScope.launch {
            (1..4).asFlow()
                .onEach {
                    delay(250)
                    if (it == 3) {
                        throw RuntimeException()
                    }
                }.onCompletion { cause ->
                    val completionThread = Thread.currentThread().name
                    if (cause != null) {
                        mTvResult.post {
                            mTvResult.append("${completionThread}:end with exception\n")
                        }
                    } else {
                        mTvResult.post {
                            mTvResult.append("${completionThread}:end normal\n")
                        }
                    }
                }.catch {//catch的调用位置决定异常捕获范围
                    val completionThread = Thread.currentThread().name
                    mTvResult.post {
                        mTvResult.append("${completionThread}:catch exception\n")
                    }
                }
//                .flowOn(Dispatchers.IO)
                .collect {//订阅，flow 才会执行
                    mTvResult.append("${Thread.currentThread().name}:$it\n")
                }
        }
    }

    /**
     * 8.retry
     */
    private fun exeFlowRetry() {
        lifecycleScope.launch {
            (1..3).asFlow()
                .onEach {
                    delay(250)
                    if (it == 3) {
                        throw RuntimeException()
                    }
                }
                .retry(2) {//等价于下面的retryWhen。须放在catch前面
                    if (it is RuntimeException) {
                        return@retry true
                    }
                    false
                }
//                .retryWhen { cause, attempt -> //attempt从0开始
//                    attempt < 2
//                }
                .catch {//catch不能放在retry前面调用，否则retry代码不执行
                    val completionThread = Thread.currentThread().name
                    mTvResult.post {
                        mTvResult.append("${completionThread}:catch exception\n")
                    }
                }
                .collect {
                    mTvResult.append("${Thread.currentThread().name}:$it\n")
                }
        }
    }


    /*
        none/missing：未指定背压策略
        error：放入Flow的异步缓存池中的数据超限了，则抛出异常。
        buffer()：Flow的异步缓存池没有固定大小，可以无限制添加数据，不会抛异常，但会OOM。
        drop：Flow的异步缓存池满了，会丢弃待入池数据。参考kotlin-flow-extensions库onBackpressurureDrop扩展
        conflate()：同drop，不同的是，会将最后一条数据强制放入缓存池中。
     */
    /**
     * 9.背压part - buffer
     */
    private fun exeBuffer() {
        //no buffer
        lifecycleScope.launch {
            var startMills = 0L
            (1..5).asFlow()
                .onEach {
                    delay(250)
                }
                //Flow的异步缓存池没有固定大小，可以无限制添加数据，不会抛异常，但会OOM。
//                .buffer()//no buffer
                .onStart {
                    startMills = System.currentTimeMillis()
                }
                .onCompletion {
                    mTvResult.append("No buffer was Done in ${System.currentTimeMillis() - startMills}ms\n")
                }
                .collect {
                    delay(100)
                    mTvResult.append("${Thread.currentThread().name}:$it\n")
                }
        }
        //use buffer
        lifecycleScope.launch {
            delay(5000)
            var startMills = 0L
            (1..5).asFlow()
                .onEach {
                    delay(250)
                }
                //Flow的异步缓存池没有固定大小，可以无限制添加数据，不会抛异常，但会OOM。
                .buffer()//注释，不使用buffer并发操作，会更慢
                .onStart {
                    startMills = System.currentTimeMillis()
                }
                .onCompletion {
                    mTvResult.append("With buffer was Done in ${System.currentTimeMillis() - startMills}ms\n")
                }
                .collect {
                    delay(100)
                    mTvResult.append("buffer ${Thread.currentThread().name}:$it\n")
                }
        }
    }


    /*
       none/missing：未指定背压策略
       error：放入Flow的异步缓存池中的数据超限了，则抛出异常。
       buffer()：Flow的异步缓存池没有固定大小，可以无限制添加数据，不会抛异常，但会OOM。
       drop：Flow的异步缓存池满了，会丢弃待入池数据。参考kotlin-flow-extensions库onBackpressurureDrop扩展
       conflate()合并：Flow的异步缓存池满了，会丢弃待入池数据，最后一条数据将强制放入缓存池中。
    */

    /**
     * 10.背压缓存侧率- 合并conflate
     */
    private fun exeConflate() {
        lifecycleScope.launch {
            //no conflate
            lifecycleScope.launch {
                var startMills = 0L
                (1..5).asFlow()
                    .onStart {
                        startMills = System.currentTimeMillis()
                    }
                    .onEach {
                        delay(100)
                    }
//                .conflate()//no buffer
                    .onCompletion {
                        mTvResult.append("No buffer was Done in ${System.currentTimeMillis() - startMills}ms\n")
                    }
                    .collect {
                        delay(500)
                        mTvResult.append("${Thread.currentThread().name}:$it\n")
                    }
            }
            //use conflate
            lifecycleScope.launch {
                delay(15000)
                var startMills = 0L
                (1..5).asFlow()
                    .onStart {
                        startMills = System.currentTimeMillis()
                    }
                    .onEach {
                        delay(100)
                        val completionThread = Thread.currentThread().name
                        mTvResult.post {
                            mTvResult.append("${completionThread}: emit: $it\n")
                        }
                    }
                    //Flow的异步缓存池满了，待入池数据会被丢弃，最后一条数据将强制放入缓存池中。
                    .conflate()//use conflate
                    .onCompletion {
                        mTvResult.append("use conflate was Done in ${System.currentTimeMillis() - startMills}ms\n")
                    }
                    .collect {
                        delay(500)
                        mTvResult.append("conflate ${Thread.currentThread().name}:$it\n")
                    }
            }
        }
    }

    /**
     * 11.并发，flatMapMerge
     */
    @FlowPreview
    private fun exeFlatMapMerge(){
        lifecycleScope.launch {
            (1..50).asFlow()
                .flatMapMerge {//flatMapMerge 是顺序调用内部代码块，并且并行地执行 collect 函数
                flow {
                    emit(it)
                }.flowOn(Dispatchers.IO)
            }.collect {
                mTvResult.append("flatMapMerge ${Thread.currentThread().name}:$it\n")
            }
        }
    }

    /**
     * 12.flatMapLatest
     */
    private fun exeFlatMapLatest(){
        lifecycleScope.launch {
            (1..5).asFlow()
                .flatMapLatest {//当发射了新值之后，上个 flow 就会被取消
                flow {
                    emit(it)
                    mTvResult.append("emit-1:$it\n")
                    delay(300)
                    mTvResult.append("emit===2:$it\n")
                }
            }.collect {
            }
        }
    }

    /**
     * 13.flatMapLatest
     */
    private fun exeFlatMapConcat(){
        lifecycleScope.launch {
            (1..5).asFlow()
                .onEach { delay(100) }
                .flatMapConcat {//collect 函数在收集新值之前会等待 flatMapConcat 内部的 当前flow 执行完毕。
                flow {
                    emit(it)
                    mTvResult.append("emit---1:$it\n")
                    delay(300)
                    mTvResult.append("emit===2:$it\n")
                }
            }.collect {
                    mTvResult.append("exeFlatMapConcat ${Thread.currentThread().name}:$it\n")
            }
        }
    }

    /**
     * 14.flattenMerge
     */
    private fun exeFlattenMerge(){
        //flattenMerge
        lifecycleScope.launch {
            val flowNum = (1..5).asFlow()
            val flowLetter = mutableListOf("a","b","c","d","e").asFlow()
            flowOf(flowNum ,flowLetter)
                .flattenMerge()// concurrency ==1 与flattenConcat一致
                .onEach { delay(100) }
                .collect{
                    mTvResult.append("flattenMerge ${Thread.currentThread().name}:$it\n")
                }
        }
        //flattenConcat
        lifecycleScope.launch {
            delay(15000)
            val flowNum = (1..5).asFlow()
            val flowLetter = mutableListOf("a","b","c","d","e").asFlow()
            flowOf(  flowNum ,flowLetter).flattenConcat()
                .onEach { delay(100) }
                .collect{
                    mTvResult.append("flattenConcat ${Thread.currentThread().name}:$it\n")
                }
        }
    }

    /**
     * 15.combine
     */
    private fun exeCombine(){
        lifecycleScope.launch {
            val flowNum = (1..4).asFlow()
            val flowLetter = mutableListOf("a","b","c","d","e").asFlow()
            //每次从 flowA 发出新的 item ，会将其与 flowB 的最新的 item 合并。
            //结果不可控
            flowNum.combine(flowLetter){ a, b ->
                "$a combine $b"
            }.collect {
                mTvResult.append("combine ${Thread.currentThread().name}:$it\n")
            }
        }
    }

    /**
     * 16.zip
     */
    private fun exeZip(){
        lifecycleScope.launch {
            val flowNum = (1..4).asFlow()
            val flowLetter = mutableListOf("a","b","c","d","e").asFlow()
            //zip对A、B进行逐个合并，并发射
            //若item数量不一致，最终结果个数==最少的那个
            flowNum.zip(flowLetter){a ,b ->
                "$a zip $b"
            }.collect {
                mTvResult.append("zip ${Thread.currentThread().name}:$it\n")
            }
        }
    }

    /**
     * 17.fold 折叠、累积操作，可设置初始值
     */
    private fun exeFold(){
        lifecycleScope.launch {
            val sum = (1..3).asFlow()
                .fold(0){ a,b ->//参数0为累加起始值
                    a + b
                }
            val multi = (1..3).asFlow()
                .fold(1){ a,b ->//参数1为累乘起始值
                    a * b
                }
            mTvResult.append("fold(0) add ${Thread.currentThread().name}:$sum\n")
            mTvResult.append("fold(1) multi ${Thread.currentThread().name}:$multi\n")
        }
    }

    /**
     * 18.reduce 合并、累积操作
     */
    private fun exeReduce(){
        lifecycleScope.launch {
            val sum = (1..3).asFlow()
                .map {
                    it * it
                }
                .reduce{ a,b ->//逐个合并
                    a + b
                }
            mTvResult.append("reduce ${Thread.currentThread().name}:$sum\n")
        }
    }

    /**
     * 19.take
     */
    private fun exeTake(){
        lifecycleScope.launch {
            (1..5).asFlow()
                .take(3)//只取前面几个emit发射的值
                .collect {
                    mTvResult.append("take ${Thread.currentThread().name}:$it\n")
                }
        }
    }

    /**
     * 20.transform
     */
    private fun exeTransform(){
       lifecycleScope.launch {
           (1..5).asFlow()
               .transform<Int,Int> { it ->
                   //transform操作可以多次调用emit,map不可以
                   emit(it )
                   emit(it * it)
               }.collect {
                   mTvResult.append("transform ${Thread.currentThread().name}:$it\n")
               }
       }
    }

}